package com.niodata.dt.fs;

import com.obs.services.ObsClient;
import com.obs.services.ObsConfiguration;
import com.obs.services.exception.ObsException;
import com.obs.services.model.AppendObjectRequest;
import com.obs.services.model.CopyObjectResult;
import com.obs.services.model.DeleteObjectResult;
import com.obs.services.model.GetObjectRequest;
import com.obs.services.model.ListObjectsRequest;
import com.obs.services.model.ObjectListing;
import com.obs.services.model.ObjectMetadata;
import com.obs.services.model.ObsObject;
import com.obs.services.model.PutObjectRequest;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;
import org.apache.commons.io.IOUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataInputStream;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FSInputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.util.Progressable;

public class ObsFileSystem extends FileSystem {

  private static final Log LOG = LogFactory.getLog(ObsFileSystem.class);

  public static final String FS_OBS_ACCESS_KEY = "fs.obs.access.key";
  public static final String FS_OBS_SECRET_KEY = "fs.obs.secret.key";
  public static final String FS_OBS_PROTOCOL =  "fs.obs.protocol";
  public static final String FS_OBS_ENDPOINT = "fs.obs.endpoint";
  public static final String FS_OBS_BUCKET = "fs.obs.bucket";
  private static final long size_128mb = 128 * 1024 * 1024;

  private String bucket;
  private String endPoint;
  private ObsClient obsClient;
  private URI uri;
  private Path workingDir;


  @Override
  public void initialize(URI name, Configuration conf) throws IOException {
    super.setConf(conf);
    super.initialize(name, conf);
    this.bucket = conf.get(FS_OBS_BUCKET);
    this.endPoint = conf.get(FS_OBS_ENDPOINT);
    //String https = conf.get(ConfigKeys.PROTOCOL, "https");
    String url = "obs://" + bucket + "." + endPoint;
    uri = URI.create(url);
    workingDir = new Path(System.getProperty("user.dir"));
    ObsConfiguration obsConf = new ObsConfiguration();
    obsConf.setConnectionTimeout(60);
    obsConf.setWriteBufferSize(4 * 1024 * 1024);
    obsConf.setSocketTimeout(60);
    obsConf.setEndPoint(endPoint);
    this.obsClient = new ObsClient(conf.get(FS_OBS_ACCESS_KEY),
      conf.get(FS_OBS_SECRET_KEY), endPoint);
  }


  @Override
  public URI getUri() {
    return this.uri;
  }

  @Override
  public FSDataInputStream open(Path f, int bufferSize) throws IOException {
    FileStatus status = getFileStatus(f);
    if (status.isDirectory()) {
      throw new IOException("file is a directory:" + f.getName());
    }
    String key = pathToKey(f);
    ObsInputStream obsInputStream = new ObsInputStream(key);
    return new FSDataInputStream(obsInputStream);
  }

  @Override
  public FSDataOutputStream create(Path f, FsPermission permission, boolean overwrite, int bufferSize,
                                   short replication, long blockSize, Progressable progress) throws IOException {
    OutToInputBufferedStream out = new OutToInputBufferedStream(true);
    FSDataOutputStream stream = new FSDataOutputStream(out, null);
    final String key = pathToKey(f);
    if (exists(f)) {
      if (!overwrite) {
        throw new IOException("file exists:" + key);
      }
      FileStatus status = getFileStatus(f);
      if (status.isDirectory()) {
        throw new IOException("exist path is directory:" + key);
      }
    }
    new Thread() {
      public void run() {
        obsClient.putObject(bucket, key, out.getInputStream());
      }
    }.start();
    return stream;
  }

  @Override
  public FSDataOutputStream append(Path f, int bufferSize, Progressable progress) throws IOException {
    if (!exists(f)) {
      return create(f);
    }
    FileStatus status = getFileStatus(f);
    String key = pathToKey(f);
    if (status.isDirectory()) {
      throw new IOException("append path is directory:" + key);
    }
    OutToInputBufferedStream out = new OutToInputBufferedStream(true);
    FSDataOutputStream stream = new FSDataOutputStream(out, null);
    new Thread() {
      public void run() {
        AppendObjectRequest req = new AppendObjectRequest();
        req.setBucketName(bucket);
        req.setObjectKey(key);
        req.setInput(out.getInputStream());
        obsClient.appendObject(req);
      }
    }.start();
    return stream;
  }

  @Override
  public boolean rename(Path src, Path dst) throws IOException {
    if (!exists(src)) {
      throw new FileNotFoundException(pathToKey(src));
    }
    if (exists(dst)) {
      throw new IOException("file exist:" + pathToKey(dst));
    }
    //check dst parent paths
    String[] parts = pathToKey(dst).split("/");
    StringBuffer sbu = new StringBuffer();
    for (int i = 0; i < parts.length; i++) {
      sbu.append("/").append(parts[i]);
      Path path = new Path(sbu.toString());
      try {
        FileStatus pathStatus = getFileStatus(path);
        if (pathStatus.isFile()) {
          throw new IOException("file exists:" + sbu.toString());
        }
      } catch (FileNotFoundException e) {
        break;
      }
    }

    FileStatus status = getFileStatus(src);
    if (status.isFile()) {
      /**
       405, ResponseStatus: Method Not Allowed, XML Error Message: <
       obs not support rename ?
        RenameRequest req = new RenameRequest();
        req.setBucketName(bucket);
        req.setObjectKey(pathToKey(src));
        req.setNewObjectKey(pathToKey(dst));
        RenameResult result = obsClient.renameFile(req);
        return result.getStatusCode() == 200;
      **/
      String srcKey = pathToKey(src);
      String dstKey = pathToKey(dst);
      CopyObjectResult result = obsClient.copyObject(bucket, srcKey, bucket, dstKey);
      obsClient.deleteObject(bucket, srcKey);
      return result.getStatusCode() == 200;
    } else {
      FileStatus[] statuses = listStatus(src);
      if (statuses.length > 0) {
        throw new IOException("directory not empty:" + src.getName());
      } else {
        /**
        RenameRequest req = new RenameRequest();
        req.setBucketName(bucket);
        req.setObjectKey(pathToKey(src) + "/");
        req.setNewObjectKey(pathToKey(dst) + "/");
        RenameResult result = obsClient.renameFolder(req);
        return result.getStatusCode() == 200;
         **/
        delete(src, false);
        mkdirs(dst);
        return true;
      }
    }
  }


  //recursive not support
  @Override
  public boolean delete(Path f, boolean recursive) throws IOException {
    if (!exists(f)) {
      return true;
    }
    FileStatus status = getFileStatus(f);
    if (status.isFile()) {
      String key = pathToKey(f);
      DeleteObjectResult result = obsClient.deleteObject(bucket, key);
      return result.getStatusCode() == 200;
    }
    FileStatus[] statuses = listStatus(f);
    if (statuses.length > 0) {
      throw new IOException("directory not empty:" + f.getName());
    }
    String key = pathToKey(f) + "/";
    DeleteObjectResult result = obsClient.deleteObject(bucket, key);
    return result.getStatusCode() == 200;
  }

  @Override
  public FileStatus[] listStatus(Path f) throws FileNotFoundException, IOException {
    String dir = pathToKey(f);
    if (!dir.endsWith("/")) {
      dir = dir + "/";
    }
    if (dir.equals("/")) {
      dir = "";
    }
    final String pathKey = dir;
    String marker = null;
    List<ObsObject> objects = new ArrayList<>();
    List<String> dirList = new ArrayList<>();
    while (true) {
      ListObjectsRequest request = new ListObjectsRequest();
      request.setBucketName(bucket);
      request.setMarker(marker);
      request.setPrefix(pathKey);
      request.setMaxKeys(5000);
      request.setDelimiter("/");

      ObjectListing listing = obsClient.listObjects(request);
      marker = listing.getNextMarker();
      objects.addAll(listing.getObjects());
      List<String> dirs = listing.getCommonPrefixes();
      dirList.addAll(dirs);
      if (marker == null) {
        break;
      }
    }
    List<FileStatus> statuses = objects.stream().filter(x -> {
      return ! x.getObjectKey().equals(pathKey);
    }).map(x -> {
      return obsObjToFileStatus(x); }).collect(Collectors.toList());
    for (String d:dirList) {
      statuses.add(new FileStatus(0, true, 1, size_128mb, 0, new Path(d)));
    }
    return statuses.toArray(new FileStatus[statuses.size()]);
  }

  @Override
  public void setWorkingDirectory(Path newDir) {
    this.workingDir = newDir;
  }

  @Override
  public Path getWorkingDirectory() {
    return this.workingDir;
  }

  @Override
  public boolean mkdirs(Path f, FsPermission permission) throws IOException {
    try {
      getFileStatus(f);
    } catch (FileNotFoundException ex) {
      String key = pathToKey(f) + "/";
      if (key.equals("/")) {
        return false;
      }
      PutObjectRequest req = new PutObjectRequest();
      req.setBucketName(bucket);
      req.setObjectKey(key);
      obsClient.putObject(req);
      return true;
    }
    return false;
  }

  @Override
  public FileStatus getFileStatus(Path f) throws IOException {
    String key = pathToKey(f);
    try {
      //check file object
      ObsObject object = obsClient.getObject(bucket, key);
      FileStatus status = obsObjToFileStatus(object);
      return status;
    } catch (ObsException obsException) {
      if (obsException.getResponseCode() == 404) {
        //check dir object
        ListObjectsRequest req = new ListObjectsRequest();
        req.setBucketName(bucket);
        req.setPrefix(key + "/");
        ObjectListing listing = obsClient.listObjects(req);
        if (listing.getObjects().size() == 0) {
          throw new FileNotFoundException(key);
        } else {
          return new FileStatus(0, true, 1, size_128mb, 0 , f);
        }
      }
      throw obsException;
    }
  }

  private FileStatus obsObjToFileStatus(ObsObject object) {
    long lastModify = 0;
    if (object.getMetadata() != null && object.getMetadata().getLastModified() != null) {
      lastModify = object.getMetadata().getLastModified().getTime();
    }
    FileStatus status;
    String key = object.getObjectKey();
    Path f = new Path(key);
    if (key.endsWith("/")) {
      status = new FileStatus(0, true, 1, size_128mb, lastModify, f);
    } else {
      long length = 0;
      if (object.getMetadata() != null) {
        length = object.getMetadata().getContentLength();
      }
      status = new FileStatus(length, false, 1, size_128mb, lastModify, f);
    }
    return status;
  }

  private String pathToKey(Path path) {
    String key = Path.getPathWithoutSchemeAndAuthority(path).toString();
    if (key.startsWith("/")) {
      key = key.substring(1);
    }
    if (key.startsWith("/")) {
      key = key.substring(1);
    }
    return key;
  }

  private class ObsInputStream extends FSInputStream {
    private long length;
    private String key;
    private int bufferSize = 16 * 1024 * 1024;
    private volatile ByteBuffer buffer;
    private volatile long pos;

    public ObsInputStream(String key) {
      this.key = key;
      ObjectMetadata metadata = obsClient.getObjectMetadata(bucket, this.key);
      length = metadata.getContentLength();
    }

    @Override
    public void seek(long pos) throws IOException {
      if (pos > length) {
        throw new IOException("end of file");
      }
      this.pos = pos;
      this.fillBuffer();
    }

    @Override
    public long getPos() throws IOException {
      return pos;
    }

    @Override
    public boolean seekToNewSource(long targetPos) throws IOException {
      return false;
    }

    @Override
    public int read() throws IOException {
      if (pos == length) {
        return -1;
      }
      if (buffer == null || buffer.remaining() == 0) {
        fillBuffer();
      }
      int b =  buffer.get();
      if (b < 0) {
        b = b + 256;
      }
      pos++;
      return b;
    }

    private void fillBuffer() throws IOException {
      if (buffer == null) {
        buffer = ByteBuffer.allocate(bufferSize);
      }
      GetObjectRequest req = new GetObjectRequest();
      req.setBucketName(bucket);
      req.setObjectKey(key);
      long end = bufferSize + pos;
      if (end > length) {
        end = length;
      }
      req.setRangeStart(pos);
      req.setRangeEnd(end);
      ObsObject object = obsClient.getObject(req);
      buffer.clear();
      InputStream input = object.getObjectContent();
      byte[] bytes = IOUtils.readFully(input, (int) (end - pos));
      input.close();
      buffer.put(bytes);
      buffer.flip();
    }
  }
}


